package dbRedis

import (
	"fmt"
	"gitee.com/kingzyt/common/log"
	. "gitee.com/kingzyt/common/objPool"
	. "gitee.com/kingzyt/common/ringlist"
	. "gitee.com/kingzyt/common/util"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/gomodule/redigo/redis"
)

/*[待改进]
有没有同步操作，决定m&e会不会与
waitGroup
syncDbOp
chan
发生死锁，
是不是在同一个goroutine中操作，决定是否能保证dbop都在m与e之间被原子化
一般都是m，userX db op ，e，但也有可能在其他goroutine发生userX的dbop，但因为是在其他goroutine中，即使sync也不会死锁，
而且副作用仅仅是加入到了当前进行中的m&e块中，被当做一个原子操作了，虽然实际上没有必要；
因为除了userActoinProc块以外的都不添加m&e，所以没有问题

db底层修改添加Multi&Exec支持，multi&exec对需要有id标记，同一个id的放入一个wait队列，当收到exec后插入发送队列；
这个功能仅作为script的特殊补充，不能大规摸使用，m&e对的内容必须是独立的，没有与其他语句有执行顺序上的要求，
不然会执行顺序不是调用的顺序；而且不能与syncOpdb联用，因为在exec执行前结果不会返回，
而内部的syncOpdb要等待结果才能执行exec，就死锁了（所以公会部分不能用m&e，公会大量使用了syncOpdb）；
我们对于新手环节上的各个步骤，服务器都要配合修改，用m&e来保证新手的步骤号与玩家游戏数据的修改是原子的（公会没有强流程新手），
m&e组内保证执行顺序与调用顺序一致，但m&e组之间无法保证执行顺序与调用顺序一致，玩家的主动发送的消息操作，除了公会部分都是可以m&e的，
对于DBAccessor中的基础接口，加入组号，默认0号直接发送，非0检测是否有multi对应组，没有就直接发，
multi出了就必须在消息结束的时候（无论成功失败panic）defer执行exec
OperateUserDB和OperateUserDBCareFail是核心的User操作db的入口，可以获得uid来做组处理，而且绕过syncOpDb，但需要严格排查一下是否还有其他的操作入口
*/

type DBOpRltProc func(ok bool, datas []interface{})

const (
	DB_rlt_fail = 0
	DB_rlt_OK   = 1
)

const (
	redis_connect_timeout = time.Second * 10
	redis_read_timeout    = time.Second * 10
	redis_write_timeout   = time.Second * 10
)

func GetInt64(value interface{}, valueName string) (rlt int64, ok bool) {
	rlt, ok = value.(int64)
	if !ok {
		log.Error("", "%s gotten from db is not int64 type, %v", valueName, value)
	}
	return
}
func GetFloat64(value interface{}, valueName string) (rlt float64, ok bool) {
	var rlt0 []byte
	rlt0, ok = value.([]byte)
	if ok {
		var err error
		rlt, err = strconv.ParseFloat(string(rlt0), 32)
		if err != nil {
			ok = false
			log.Error("", "%s gotten from db is not float64 type, %v", valueName, value)
		}
	} else {
		log.Error("", "%s gotten from db is not float64 type, %v", valueName, value)
	}
	return
}
func GetBytes(value interface{}, valueName string) (rlt []byte, ok bool) {
	rlt, ok = value.([]byte)
	if !ok {
		log.Error("", "%s gotten from db is not []byte type, %v", valueName, value)
	}
	return
}
func GetValues(value interface{}, valueName string) (rlt []interface{}, ok bool) {
	rlt, ok = value.([]interface{})
	if !ok {
		log.Error("", "%s gotten from db is not []interface{} type, %v", valueName, value)
	}
	return
}

func GetPRecover(module string, ok *bool) {
	if r := recover(); r != nil {
		log.Error(module, "%s", r)
		PrintPanicTrace()
		if ok != nil {
			*ok = false
		}
	}
}
func GetInt64P(value interface{}, valueName string) (rlt int64) {
	rlt, ok := value.(int64)
	if !ok {
		panic(fmt.Errorf("%s gotten from db is not int64 type, %v", valueName, value))
	}
	return
}
func GetFloat64P(value interface{}, valueName string) (rlt float64) {
	rlt0, ok := value.([]byte)
	if ok {
		var err error
		rlt, err = strconv.ParseFloat(string(rlt0), 32)
		if err != nil {
			panic(fmt.Errorf("%s gotten from db is not float64 type, %v", valueName, value))
		}
	} else {
		panic(fmt.Errorf("%s gotten from db is not float64 type, %v", valueName, value))
	}
	return
}
func GetBytesP(value interface{}, valueName string) (rlt []byte) {
	rlt, ok := value.([]byte)
	if !ok {
		panic(fmt.Errorf("%s gotten from db is not []byte type, %v", valueName, value))
	}
	return
}
func GetValuesP(value interface{}, valueName string) (rlt []interface{}) {
	rlt, ok := value.([]interface{})
	if !ok {
		panic(fmt.Errorf("%s gotten from db is not []interface{} type, %v", valueName, value))
	}
	return
}

type DBScript struct {
	name           string
	keyCount       int
	scriptStr      string
	hintParamsSize int

	//loaded bool

	// Script obj load(do at DBAccessor.Start) & send(do at DBAccessor.sendDBCommands) will never do at same time, DBAccessor promises it
	// ref to the Script obj is no change in whole time
	script *redis.Script
}

func NewDBScript(name string, keyCount int, hintParamsSize int, scriptStr string) *DBScript {
	rlt := &DBScript{name: name, keyCount: keyCount, hintParamsSize: hintParamsSize, scriptStr: scriptStr}
	rlt.script = redis.NewScript(rlt.keyCount, rlt.scriptStr)
	return rlt
}
func (self *DBScript) GetName() string {
	return self.name
}
func (self *DBScript) GetScriptContent() string {
	return self.scriptStr
}

// one game action do one db update/get
type DBAction struct {
	script  *DBScript
	args    []interface{}
	rltProc DBOpRltProc
	rltOk   bool
	datas   []interface{}
}

func (self *DBAction) safeRltProc() {
	if self.rltProc != nil {
		defer func() {
			if r := recover(); r != nil {
				log.Error(self.script.name, "panic happen:%s", r)
				PrintPanicTrace()
			}
		}()

		self.rltProc(self.rltOk, self.datas)
	}
}

func (self *DBAction) clean() {
	self.script = nil
	self.args = nil
	self.rltProc = nil
	self.rltOk = false
	self.datas = nil
}

const (
	status_stopped int32 = iota
	status_running
	status_stopping
)

type DBAccessor struct {
	name       string
	dbAddr     string
	dbPassword string
	dbConn     redis.Conn

	rootMutex sync.Mutex

	recycleLine    *ObjPool
	sendLine       chan *DBAction
	flushLine      chan *DBAction
	receiveLine    chan *DBAction
	procLine       *RingList
	procLineMutex  sync.Mutex
	procLineN      int64
	procLineNMutex sync.Mutex
	procLineNCond  *sync.Cond

	stopSign chan byte
	endWg    sync.WaitGroup
	status   int32

	scripts []*DBScript
}

const default_max_db_pile_up_command_count = 1000
const default_max_flush_size = 0 // directly send seems faster..., need tune at real env
const default_force_flush_time_interval = time.Millisecond * 50
const default_fix_db_link_time_interval = time.Second * 5

func NewAccessor(name string, maxPileUpCmdCount int) *DBAccessor {
	if maxPileUpCmdCount <= 0 {
		maxPileUpCmdCount = default_max_db_pile_up_command_count
	}

	rlt := &DBAccessor{
		name: name,

		/*
			如果recycleLine像其他Line一样使用chan的话,当在dbop的回调中调用dbop时,
			在高速读写的时候,有几率会造成chan阻塞,而且因为dbop回调处理是所有回调统一在一个单goroutine中进行的,
			一旦阻塞,就会造成receiveLine的ac无法进入recycleLine,造成之后的dbop都阻塞,
			于是第一个阻塞的dbop阻塞在处理环节,而其余所有的dbop都阻塞在dbop调用.
			所以为了解决dbop中嵌套dbop的问题,recycleLine采用无阻塞的objPool
		*/
		recycleLine: NewObjPool(name+"_recycleLine", 20, 10, func() interface{} {
			return &DBAction{}
		}),
		sendLine:    make(chan *DBAction, maxPileUpCmdCount),
		flushLine:   make(chan *DBAction, maxPileUpCmdCount),
		receiveLine: make(chan *DBAction, maxPileUpCmdCount),
		procLine:    NewRingList(nil, maxPileUpCmdCount),

		stopSign: make(chan byte, 2),
		scripts:  make([]*DBScript, 0),
	}

	rlt.procLineNCond = sync.NewCond(&rlt.procLineNMutex)

	// we only donot stop procRlt goroutine, no redis db calls, just proc results
	go rlt.procDBOpResult()

	return rlt
}

func (self *DBAccessor) GetName() string {
	return self.name
}

// 这里缓存script, 为的是在db link中断后的重启时,能够自动修复script的使用
func (self *DBAccessor) AddDBScript(script *DBScript) {
	//log.Info(self.name, log.Trace, "---- %s -----------------", script.name)
	//log.Info(self.name, log.Trace, script.scriptStr)

	self.rootMutex.Lock()
	defer self.rootMutex.Unlock()

	for _, scr := range self.scripts {
		if scr == script {
			return
		}
	}
	self.scripts = append(self.scripts, script)
}

func (self *DBAccessor) ShutdownDB() {
	self.dbConn.Do("SHUTDOWN")
	wg := sync.WaitGroup{}
	wg.Add(1)
	self.stop("shutdown", false, func() {
		wg.Done()
	})
	wg.Wait()
}

func (self *DBAccessor) Quit() {
	wg := sync.WaitGroup{}
	wg.Add(1)
	self.stop("quit", false, func() {
		wg.Done()
	})
	wg.Wait()
}

// no quit
func (self *DBAccessor) Start(addr string, password string) bool {
	self.rootMutex.Lock()
	defer self.rootMutex.Unlock()

	if atomic.LoadInt32(&self.status) == status_running {
		return true
	}

	// stop by the err, maybe leave some stop sign in chan, clear them
	for i := 0; i < len(self.stopSign); i++ {
		_ = <-self.stopSign
	}
	log.Info(self.name, log.I, "stopSign len:%d", len(self.stopSign))

	self.recycleAllFailAC()

	self.dbAddr = MergeSameMachineServerIPToLocal(addr)
	conn, err := redis.Dial("tcp", self.dbAddr, redis.DialConnectTimeout(redis_connect_timeout), redis.DialReadTimeout(redis_read_timeout), redis.DialWriteTimeout(redis_write_timeout))
	if err != nil {
		log.Error(self.name, "db connect fail, err:%s", err)
		return false
	}

	self.dbConn = conn

	if password != "" {
		self.dbPassword = password
		_, err = self.dbConn.Do("AUTH", self.dbPassword)
		if err != nil {
			log.Error(self.name, "db auth fail, err:%s", err)
			return false
		}
	}

	// check until the db is  ready for service
	dbSize, err := redis.Int(self.dbConn.Do("DBSIZE"))
	if err != nil {
		if strings.Contains(err.Error(), "LOADING Redis is loading the dataset in memory") {
			checkDBOKTimer := time.NewTicker(time.Second).C
			for {
				<-checkDBOKTimer
				dbSize, err = redis.Int(self.dbConn.Do("DBSIZE"))
				if err == nil {
					break
				}
			}
		} else {
			log.Error(self.name, "db check fail, err:%s", err)
			return false
		}
	}
	log.Info(self.name, log.I, "db(%s) total key count:%d", self.dbAddr, dbSize)

	for _, s := range self.scripts {
		err := self.ScriptLoad(s)
		if err != nil {
			panic(err)
		}
	}

	if !atomic.CompareAndSwapInt32(&self.status, status_stopped, status_running) {
		panic(fmt.Errorf("%s, need status_stopped, but %d", self.name, atomic.LoadInt32(&self.status)))
	}

	go self.sendDBCommands()
	go self.receiveDBOpResult()

	return true
}

func (self *DBAccessor) ScriptLoad(script *DBScript) error {
	err := script.script.Load(self.dbConn)
	if err != nil {
		return fmt.Errorf("%s, script(%s) load err:%s", self.name, script.name, err)
	}

	hash := script.script.Hash()
	log.Info(self.name, log.I, "load script [%s] ok, %s", script.GetName(), hash)
	return nil
}

func (self *DBAccessor) ShowOpGoingCnt() {
	procLineLen := 0
	func() {
		self.procLineMutex.Lock()
		defer self.procLineMutex.Unlock()
		procLineLen = self.procLine.Len()
	}()
	log.Info(self.GetName(), log.Trace, "DBGoing send %d, flush %d, receive %d, proc %d, recycle %d", len(self.sendLine), len(self.flushLine), len(self.receiveLine), procLineLen, self.recycleLine.GetCurObjCnt())
}

func (self *DBAccessor) addToProcLine(ac *DBAction, rltOk bool, datas []interface{}) {
	ac.rltOk = rltOk
	ac.datas = datas

	// 这里的procLineMutex嵌套procLineNMutex, 只是做了一个保证: [将ac取出procLine]不会发生在[ac加入procLine]与[N++]之间,只能是之前或之后
	self.procLineMutex.Lock()
	defer self.procLineMutex.Unlock()
	self.procLine.PushTail(ac)

	self.procLineNMutex.Lock()
	defer self.procLineNMutex.Unlock()
	self.procLineN++
	self.procLineNCond.Signal()
}

func (self *DBAccessor) recycleAllFailAC() {
	var ac *DBAction
	/*
		虽然sendLine还没有发送，尚未进行任何DBAccessor相关操作，但如果搁置等待DBAccessor修复的话，
		必然造成上游C的请求超时重试，所以我们直接清空所有的line到recycleline中，
		并且执行所有的调用方的失败处理，尽快告知上游出错
	*/
	lenSendChan := len(self.sendLine)
	for i := 0; i < lenSendChan; i++ {
		ac = <-self.sendLine
		self.addToProcLine(ac, false, nil)
	}

	lenFlushChan := len(self.flushLine)
	for i := 0; i < lenFlushChan; i++ {
		ac = <-self.flushLine
		self.addToProcLine(ac, false, nil)
	}

	lenReceiveChan := len(self.receiveLine)
	for i := 0; i < lenReceiveChan; i++ {
		ac = <-self.receiveLine
		self.addToProcLine(ac, false, nil)
	}
}

/*
与DBAccessor发生的任何错误都是严重错误，包括Send、Flush、Receive造成的G端消息处理超时，都需要玩家重新登录，触发G的重新载入玩家数据
因为G没有直接的手段获知DBAccessor处理消息出错，需要加入errCallback，告诉G出错，同时每一个DBAccessor指令都要加uid或是其他标示，出错时可以返回给errCallback
*/
func (self *DBAccessor) stop(reason string, keepTryLink bool, onStop func()) bool {
	// make sure stop() will not be in by two goroutine at same time
	self.rootMutex.Lock()
	defer self.rootMutex.Unlock()

	if atomic.LoadInt32(&self.status) == status_stopping || atomic.LoadInt32(&self.status) == status_stopped {
		return true
	}

	if !atomic.CompareAndSwapInt32(&self.status, status_running, status_stopping) {
		panic(fmt.Errorf("%s, need status_running, but %d", self.name, atomic.LoadInt32(&self.status)))
	}

	log.Info(self.name, log.I, "DBAccessor accessor stop reason:%s", reason)

	self.endWg.Add(2)
	self.stopSign <- 1
	self.stopSign <- 1

	go func() {
		func() {
			// redis.Conn hold lock, maybe some case lock, should be not warpped by lock; AND we should call close before signal wait, to cancel conn read/write wait
			self.dbConn.Close()

			log.Info(self.name, log.I, "stop wait all stop sign......")
			self.endWg.Wait()
			log.Info(self.name, log.I, "all stop sign ok")

			self.rootMutex.Lock()
			defer self.rootMutex.Unlock()

			// new chan, clear all left sign
			self.stopSign = make(chan byte, 2)

			self.recycleAllFailAC()

			if !atomic.CompareAndSwapInt32(&self.status, status_stopping, status_stopped) {
				panic(fmt.Errorf("%s, need status_stopping, but %d", self.name, atomic.LoadInt32(&self.status)))
			}

			if onStop != nil {
				onStop()
			}
		}()

		if keepTryLink {
			log.Warn(self.name, "DB link lost, try to fix after %s", default_fix_db_link_time_interval)
			// retry to link db forever

			c := time.NewTicker(default_fix_db_link_time_interval).C
			for {
				<-c
				if self.Start(self.dbAddr, self.dbPassword) {
					log.Info(self.name, log.I, "DB link fix OK")
					break
				}
			}
		}
	}()

	return true
}

func (self *DBAccessor) sendDBCommands() {
	defer func() {
		log.Info(self.name, log.I, "sendDBCommands stop sign ok")
		self.endWg.Done()
	}()

	flushSizeSum := 0

	flush := func() bool {
		waitingFlushCount := len(self.flushLine)
		if waitingFlushCount > 0 {
			err := self.dbConn.Flush()
			// fatal err
			if err != nil {
				self.stop(fmt.Sprintf("Flush err: %s", err), true, nil)
				self.dbLinkErrProc()
				log.Info(self.name, log.I, "flush return false")
				return false //here make outside caller exit, ignore the stop sign
			}
			for i := 0; i < waitingFlushCount; i++ {
				ac, ok := <-self.flushLine
				if !ok {
					//here make outside caller exit, ignore the stop sign
					return false
				}

				self.receiveLine <- ac
			}
		}
		flushSizeSum = 0

		return true
	}

	timer := time.NewTicker(default_force_flush_time_interval)
	defer timer.Stop()

End:
	for {
		select {
		case <-self.stopSign:
			log.Info(self.name, log.I, "sendDBCommands: receive stop sign")
			break End

		case <-timer.C:
			if !flush() {
				break End
			}

		case ac, ok := <-self.sendLine:
			if !ok {
				break End
			}

			err := ac.script.script.SendHash(self.dbConn, ac.args...)
			// fatal err, here not stop, flush will err again, then stop
			if err != nil {
				log.Error(self.name, "Send err:%s", err)
				// 从chan中取出后出错的ac，插回procline
				self.addToProcLine(ac, false, nil)
				break
			}
			self.flushLine <- ac

			flushSizeSum += ac.script.hintParamsSize
			if flushSizeSum >= default_max_flush_size {
				if !flush() {
					break End
				}
			}
		}
	}
}

func (self *DBAccessor) receiveDBOpResult() {
	defer func() {
		log.Info(self.name, log.I, "receiveDBOpResult stop sign ok")
		self.endWg.Done()
	}()

	const (
		rlt_ac_err int = iota
		rlt_ac_fail
		rlt_ac_ok
		rlt_exit
	)

	var scriptName string
	var datas []interface{}
End:
	for {
		select {
		case ac, ok := <-self.receiveLine:
			if !ok {
				break End
			}

			rltCode := func() (rltCode int) {
				scriptName = ac.script.GetName()
				datas = nil

				rlt, err := self.dbConn.Receive()
				// fatal err
				if err != nil {
					errInfo := fmt.Sprintf("script:%s, Receive err: %s", scriptName, err)
					if self.dbConn.Err() != nil {
						self.stop(errInfo, true, nil)
						self.dbLinkErrProc()
						log.Info(self.name, log.I, "Receive happen net connect err to return")
						return rlt_exit //here make goroutine exit, ignore the stop sign
					} else {
						log.Error(self.name, errInfo)
						return rlt_ac_err
					}
				}
				/*
					dbscript中的返回都为{1/0, ...}
					无返回的:
					成功:{1}
					失败:{0, reasonString}
					有返回的:
					成功:{1, ...}
					失败:{0, reasonString, ...}
				*/
				var rltInt int64
				datas, _ = redis.Values(rlt, nil)
				if len(datas) == 0 {
					log.Error(self.name, "op %s, result format err, datasLen:%d, should be >= 1", scriptName, len(datas))
					return rlt_ac_err
				}

				rltInt, ok = datas[0].(int64)
				if !ok {
					log.Error(self.name, "op %s, result format err, first is success flag; now:[%v]", scriptName, datas[0])
					return rlt_ac_err
				}

				if rltInt == DB_rlt_OK {
					return rlt_ac_ok
				} else {
					if len(datas) < 2 {
						log.Error(self.name, "op %s, Fail result format err, first is 0, second is fail reason; now:[%v]", scriptName, datas)
						return rlt_ac_err
					}

					failReason, ok := datas[1].([]byte)
					if !ok {
						log.Error(self.name, "op %s, failReason format err, should be []byte(string); now:[%v]", scriptName, datas[1])
						return rlt_ac_err
					}
					log.Error(self.name, "op %s, fail, reason: %s", scriptName, string(failReason))
					return rlt_ac_fail
				}
			}()
			switch rltCode {
			case rlt_ac_err:
				self.addToProcLine(ac, false, nil)
			case rlt_ac_fail:
				self.addToProcLine(ac, false, datas[2:])
			case rlt_ac_ok:
				self.addToProcLine(ac, true, datas[1:])
			case rlt_exit:
				self.addToProcLine(ac, false, nil)
				break End
			}

		case <-self.stopSign:
			log.Info(self.name, log.I, "receiveDBOpResult: receive stop sign")
			break End
		}
	}
}

func (self *DBAccessor) procDBOpResult() {
	var acObj interface{}
	var ac *DBAction
	var doCnt = int64(0)
	for {
		func() {
			self.procLineMutex.Lock()
			defer self.procLineMutex.Unlock()
			acObj = self.procLine.PopHead()
		}()
		if acObj != nil {
			ac = acObj.(*DBAction)
			if ac != nil {
				ac.safeRltProc()
				ac.clean()
				self.recycleLine.Push(ac)
				ac = nil
			}
			acObj = nil
			doCnt++
		}
		func() {
			self.procLineNMutex.Lock()
			defer self.procLineNMutex.Unlock()

			self.procLineN -= doCnt
			doCnt = 0
			if self.procLineN <= 0 {
				self.procLineNCond.Wait()
			}
		}()
	}
}

/*
opdb，在stop的瞬间可能有一些op越过了状态判定，而且在stoped了之后才加入send line，
那么：1.之后db及时连接成功，那么被及时执行，没有任何问题。
2.db或dbaccessor直接停止，系统重启，造成请求被忽略，玩家被要求重新登录，也没有问题。
3.db在比较长的时间后才连接成功，在client响应超时之前，
情况同1，在超时之后，若处理了到db了，就是ok了，client不超时就等待gr反馈，超时就重登陆，从db中获得新数据。
*/

func (self *DBAccessor) OperateDBAndProcWithoutLinkErr(script *DBScript, rltProc DBOpRltProc, args ...interface{}) (ok bool) {
	if script == nil {
		panic(fmt.Errorf("%s, operateDB script should not be nil !", self.name))
	}

	if rltProc == nil {
		panic(fmt.Errorf("%s, operateDB [%s] rltProc should not be nil !", self.name, script.name))
	}

	if atomic.LoadInt32(&self.status) != status_running {
		return false
	}

	ac := self.recycleLine.Pull().(*DBAction)
	ac.script = script
	ac.args = args
	ac.rltProc = rltProc

	/*
		如果没有不限量压入的procLine, 则处理是在receive时同步处理的,
		一旦处理速度慢，造成receiveLine堆积，压力receiveLine->flushLine->sendline，
		sendline堆积，阻塞，然后在receive处理中有OperateDB send，首位咬合，死锁.
		所以要引入无阻塞的procLine来拆解.

		procLine的大前提是send的顺序就是处理的顺序,所以原则上不能分多个并发队列.
		需要将队列中的大处理量的操作(比如userload)独立go处理，尽可能保证快速消灭receive来的任务, 不卡db.
		虽然可以在游戏的角度, 按不同user或是逻辑单元来区分处理队列,
		但这样逻辑就过于复杂, 而且超过16线的并发也没有什么大意义, 这里就简单处理,
		在游戏逻辑部分只要注意不要在db成功之后的处理中做太大的事情就没有问题。

		不将上游的chan改成ringlist，因为send line只能为chan，因为可以并发插入，不是互斥锁，
		send，flush和receive chan在goroutinue之间有chan阻塞通知睡眠的降功耗作用。
		receiveLine不能替换为ringlist，虽然加一个line也是有消耗，延长了流水线，
		但receive是接受db回应的环节，这里不能挨个慢慢来，必须快速响应后交由后续处理，
		然后它自己快速清理db的堆积响应，Receive是阻塞调用卡goroutinue的，所以要保证不影响redis db的流水线
	*/
	// 插入sendline的ac不能出错，因为没有close sendline的地方，那么若非bug，不可能插入失败
	self.sendLine <- ac
	return true
}

func (self *DBAccessor) dbLinkErrProc() {
	log.Error(self.name, "dbLinkErr: %s", self.dbAddr)
}

func (self *DBAccessor) OperateDBWrap(script *DBScript, onOK func(datas []interface{}), onLinkFail func(), onLogicFail func(), args ...interface{}) {
	ok := self.OperateDBAndProcWithoutLinkErr(script, func(ok bool, datas []interface{}) {
		if !ok {
			log.Error(self.name, "db op fail, [%s]: logic err", script.name)
			if onLogicFail != nil {
				onLogicFail()
			}
			return
		}

		if onOK != nil {
			onOK(datas)
		}
	}, args...)
	if !ok {
		log.Error(self.name, "db op fail, [%s]: link err", script.name)
		if onLinkFail != nil {
			onLinkFail()
		}
	}
}

type DBErr int32

const (
	DBErr_None DBErr = iota
	DBErr_link
	DBErr_logic
)

func (self *DBAccessor) OperateDBSimpleCareFail(script *DBScript, proc func(err DBErr), args ...interface{}) {
	if proc == nil {
		panic("OperateDBSimpleCareFail, proc is nil")
	}

	ok := self.OperateDBAndProcWithoutLinkErr(script,
		func(ok bool, datas []interface{}) {
			if !ok {
				proc(DBErr_logic)
				return
			}

			proc(DBErr_None)
		}, args...)
	if !ok {
		proc(DBErr_link)
		return
	}
}

func (self *DBAccessor) SyncDoScript(module string, script *DBScript, proc func(datas []interface{}), args ...interface{}) (err error) {
	signal := make(chan byte, 1)
	ok := false
	self.OperateDBWrap(script,
		func(datas []interface{}) {
			defer func() {
				if ok {
					signal <- 1
				} else {
					signal <- 0
				}
			}()
			defer GetPRecover(script.GetName(), &ok)

			if proc != nil {
				proc(datas)
			}
			ok = true
		},
		func() { signal <- 0 },
		func() { signal <- 0 },
		args...)

	WaitSignalProc(time.Second*10, signal,
		func(val byte) {
			if val == 0 {
				err = fmt.Errorf(script.GetName() + " db fail")
				log.Error(module, err.Error())
			}
		}, func() {
			err = fmt.Errorf(script.GetName() + " db timeout")
			log.Error(module, err.Error())
		})
	return
}

func (self *DBAccessor) SyncOperateDB(module string, script *DBScript, signal chan byte, proc func(datas []interface{}), args ...interface{}) (ok bool) {
	self.OperateDBWrap(script,
		func(datas []interface{}) {
			defer func() {
				if ok {
					signal <- 1
				} else {
					signal <- 0
				}
			}()
			defer GetPRecover(script.GetName(), &ok)

			if proc != nil {
				proc(datas)
			}
			ok = true
		},
		func() { signal <- 0 },
		func() { signal <- 0 },
		args...)

	WaitSignalProc(time.Second*10, signal,
		func(val byte) {
			if val == 0 {
				ok = false
				log.Error(module, script.GetName()+" db fail")
			} else {
				ok = true
			}
		}, func() {
			ok = false
			log.Error(module, script.GetName()+" db timeout")
		})
	return
}

// script: return {1, cursor, batchCustomData...}
func SyncIteratorLargeDatasByBatchFromDB(module string, dbac *DBAccessor, script *DBScript, procEachBatch func(batchData []interface{})) bool {
	const (
		fail    = 0
		success = 1
		allend  = 2
	)

	log.Info(module, log.I, script.GetName()+" begin")

	cursor := int64(0)
	stepSign := make(chan int, 1)
	for {
		ok := dbac.OperateDBAndProcWithoutLinkErr(script,
			func(ok bool, datas []interface{}) {
				stepRlt := fail
				defer func() {
					if !ok {
						stepRlt = fail
					}
					stepSign <- stepRlt
				}()
				if !ok {
					return
				}
				defer GetPRecover(module, &ok)

				cursor = GetInt64P(datas[0], "nextCursor")
				if len(datas) > 1 {
					procEachBatch(datas[1:])
				}
				if cursor <= 0 {
					stepRlt = allend
					return
				}

				stepRlt = success
			}, cursor)
		if !ok {
			log.Info(module, log.I, script.GetName()+" db link fail")
			return false
		}

		timeout := time.NewTimer(time.Second * 10)
		select {
		case stepRlt := <-stepSign:
			timeout.Stop()
			switch stepRlt {
			case fail:
				log.Info(module, log.I, script.GetName()+" fail")
				return false
			case success:
				// continue to next batch
			case allend:
				log.Info(module, log.I, script.GetName()+" ok")
				return true
			}
		case <-timeout.C:
			timeout.Stop()
			log.Error(module, script.GetName()+" timeout")
			return false
		}
	}
}
